package parallels.servertech.server.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;

import parallels.servertech.server.nio.event.ChannelEvent;
import parallels.servertech.server.nio.event.ChannelEvent.ChannelEventType;
import parallels.servertech.server.nio.event.ChannelEventDispatcher;
import parallels.servertech.server.nio.event.CommandDecoder;

public class ChannelSelector {
	private static final Logger log = Logger.getLogger("selector");
	
	private static final int SELECTION_TIMEOUT = 1000;
	
	private final int port;
	private final Selector selector;
	private final Thread selectionProcess;
	
	private ChannelEventDispatcher dispatcher;
	
	private volatile boolean started = false;
	
	private class SelectionProcess implements Runnable {
		@Override
		public void run() {
			log.info("started");
			
			try {
				ServerSocketChannel serverChannel = ServerSocketChannel.open();
				serverChannel.bind(new InetSocketAddress(port));
				serverChannel.configureBlocking(false);
				
				serverChannel.register(selector, SelectionKey.OP_ACCEPT);
				
				while (started) {
					int selected = selector.select(SELECTION_TIMEOUT);
					if (selected > 0) {
						log.info(selected + " channels selected");
						for (Iterator<SelectionKey> keysIter = selector.selectedKeys().iterator();keysIter.hasNext();) {
							SelectionKey selKey = keysIter.next();
							if (selKey.isAcceptable()) {
								SocketChannel channel = ((ServerSocketChannel)selKey.channel()).accept();
								log.info("client " + channel.getRemoteAddress() + " channel connected");
								channel.configureBlocking(false);
								SelectionKey clientSelKey = channel.register(selector, SelectionKey.OP_READ);
								// TODO should be parameterized
								dispatcher.addEventHandler(clientSelKey, new CommandDecoder());
							} else if (selKey.isReadable()) {
								SocketChannel channel = (SocketChannel) selKey.channel();
								log.info("channel @" + channel.hashCode() + " is ready for reading");
								dispatcher.sendEvent(new ChannelEvent(dispatcher, ChannelEventType.CHANNEL_READY_FOR_READING, channel, selKey));
							} else if (selKey.isWritable()) {
								SocketChannel channel = (SocketChannel) selKey.channel();
								log.info("channel @" + channel.hashCode() + " is ready for writing");
								dispatcher.sendEvent(new ChannelEvent(dispatcher, ChannelEventType.CHANNEL_READY_FOR_WRITING, channel, selKey));
							}
							keysIter.remove();
						}
					}
				}
			} catch (IOException ioe) {
				log.log(Level.SEVERE, "receiver error: " + ioe.getClass() + "(" + ioe.getMessage() + ")");
			}
		}
	}
	
	public ChannelSelector(int port) {
		this.port = port;
		this.selectionProcess = new Thread(new SelectionProcess(), "channel-selector");
		
		try {
			this.selector = Selector.open();
		} catch (IOException ioe) {
			throw new IllegalStateException("failed to open selector: ", ioe);
		}
	}

	public void start() {
		started = true;
		dispatcher = new ChannelEventDispatcher(this);
		dispatcher.start();
		selectionProcess.start();
		log.info("channel selector started");
	}

	public void stop() {
		started = false;
		selector.wakeup();
		log.info("channel selector stopped");
	}
}
